package com.jorado.etl.repository;

import com.jorado.elasticsearch.ElasticSearchClient;
import com.jorado.elasticsearch.ElasticSearchClientFactory;
import com.jorado.elasticsearch.model.ESCluster;
import com.jorado.elasticsearch.model.ESConfiguration;
import com.jorado.elasticsearch.model.ESIndexType;
import com.jorado.search.core.service.Exporter;
import com.jorado.search.core.util.util.StringUtils;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.index.query.MatchAllQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.context.annotation.Description;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Description("es数据库职位导出器")
public class ESPositionRepository implements Exporter {

    private ESIndexType esIndexType;

    private ElasticSearchClient client;

    public ESPositionRepository(String businessKey) {

        ESConfiguration esConfiguration = null;
        if (esConfiguration != null && esConfiguration.getIndexTypes() != null && esConfiguration.getIndexTypes().size() > 0) {
            esIndexType = esConfiguration.getIndexTypes().containsKey(businessKey) ? esConfiguration.getIndexTypes().get(businessKey) : new ESIndexType();
            ESCluster esCluster = StringUtils.isNullOrWhiteSpace(esIndexType.getEsCluster()) ? null : esConfiguration.getESCluster(esIndexType.getEsCluster());
            client = initClient(true, esCluster);
        }
    }

    @Override
    public long count() {

        MatchAllQueryBuilder matchAllQueryBuilder = QueryBuilders.matchAllQuery();

        SearchRequestBuilder requestBuilder = client.getClient().prepareSearch();

        requestBuilder.setQuery(matchAllQueryBuilder);

        requestBuilder.setTypes(esIndexType.getType());
        requestBuilder.setIndices(esIndexType.getIndex());
        requestBuilder.setSearchType(SearchType.QUERY_AND_FETCH);
        requestBuilder.setSize(0);

        SearchResponse response = requestBuilder.get();

        return response.getHits().getTotalHits();
    }

    @Override
    public List<Map<String, Object>> listDatas(int start, int rows) {

        MatchAllQueryBuilder matchAllQueryBuilder = QueryBuilders.matchAllQuery();

        SearchRequestBuilder requestBuilder = client.getClient().prepareSearch();

        requestBuilder.setQuery(matchAllQueryBuilder);

        requestBuilder.setTypes(esIndexType.getType());
        requestBuilder.setIndices(esIndexType.getIndex());
        requestBuilder.setSearchType(SearchType.QUERY_AND_FETCH);
        requestBuilder.addSort("date_created", SortOrder.DESC);
        requestBuilder.setFrom(start);
        requestBuilder.setSize(rows);

        SearchResponse response = requestBuilder.get();
        List<Map<String, Object>> result = new ArrayList<>();
        for (SearchHit hit : response.getHits().getHits()) {

            result.add(hit.getSourceAsMap());
        }
        return result;
    }

    public Map<String, Object> get(String jobNumber) {

        SearchRequestBuilder requestBuilder = client.getClient().prepareSearch();
        TermQueryBuilder queryBuilder = QueryBuilders.termQuery("job_num", jobNumber);
        requestBuilder.setQuery(queryBuilder);
        requestBuilder.setTypes(esIndexType.getType());
        requestBuilder.setIndices(esIndexType.getIndex());
        requestBuilder.setSearchType(SearchType.QUERY_AND_FETCH);
        requestBuilder.setFrom(0);
        requestBuilder.setSize(1);

        SearchResponse response = requestBuilder.get();
        Map<String, Object> result = new HashMap<>();
        for (SearchHit hit : response.getHits().getHits()) {

            result = hit.getSourceAsMap();
        }
        return result;
    }

    public void insert(List<Map<String, Object>> datas) {

        for (Map<String, Object> data : datas) {
            client.index(esIndexType.getIndex(), esIndexType.getType(), data);
        }

    }

    private ElasticSearchClient initClient(boolean usePools, ESCluster esCluster) {
        return ElasticSearchClientFactory.createESClient(usePools, esCluster);
    }
}

